package com.example.service;

import com.example.loader.HazelcastLoader;
import com.hazelcast.cluster.Member;
import com.hazelcast.cluster.MembershipListener;
import com.hazelcast.collection.IQueue;
import com.hazelcast.collection.ISet;
import com.hazelcast.config.Config;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IExecutorService;
import com.hazelcast.core.LifecycleListener;
import com.hazelcast.cp.CPSubsystem;
import com.hazelcast.cp.IAtomicLong;
import com.hazelcast.cp.IAtomicReference;
import com.hazelcast.cp.ICountDownLatch;
import com.hazelcast.cp.lock.FencedLock;
import com.hazelcast.crdt.pncounter.PNCounter;
import com.hazelcast.durableexecutor.DurableExecutorService;
import com.hazelcast.map.IMap;
import com.hazelcast.multimap.MultiMap;
import com.hazelcast.replicatedmap.ReplicatedMap;
import com.hazelcast.scheduledexecutor.IScheduledExecutorService;
import com.hazelcast.topic.ITopic;
import com.hazelcast.transaction.TransactionContext;
import com.hazelcast.transaction.TransactionOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

@Service
public class HazelcastServiceImpl implements Serializable, HazelcastService, InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    private HazelcastInstance hazelcastInstance;

    @Resource
    private List<HazelcastLoader> hazelcastLoaders = new ArrayList<>();

    /***
     *
     * 初始化配置信息
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        Config config = new Config();
        for (HazelcastLoader hazelcastLoader : this.hazelcastLoaders) {
            hazelcastLoader.loadConfig(config);
        }
        this.hazelcastInstance = Hazelcast.newHazelcastInstance(config);
        for (HazelcastLoader hazelcastLoader : this.hazelcastLoaders) {
            hazelcastLoader.init(this.hazelcastInstance);
        }
    }

    /***
     * 获取应用实例
     * @return
     */
    @Override
    public HazelcastInstance getHazelcastInstance() {
        return this.hazelcastInstance;
    }


    /***
     * 获取集群对象
     * @return
     */
    @Override
    public Set<Member> getMember() {
        return this.hazelcastInstance.getCluster().getMembers();
    }


    /***
     * 节点启动监听
     * @param membershipListener
     */
    @Override
    public void addMembershipListener(MembershipListener membershipListener) {
        this.hazelcastInstance.getCluster().addMembershipListener(membershipListener);
    }


    /****
     * 添加生命周期事件监听
     * @param lifecycleListener
     */
    @Override
    public void addLifecycleListener(LifecycleListener lifecycleListener) {
        this.hazelcastInstance.getLifecycleService().addLifecycleListener(lifecycleListener);
    }


    /***
     * 生成分布式id
     * @param name
     * @return
     */
    @Override
    public long getFlakeId(String name) {
        return this.hazelcastInstance.getFlakeIdGenerator(name).newId();
    }


    /***
     * 生成CP系统操作类
     * @return
     */
    @Override
    public CPSubsystem getCPSubsystem() {
        return this.hazelcastInstance.getCPSubsystem();
    }


    /****
     * 获取Topic信息
     * @param name
     * @return
     */
    @Override
    public ITopic getTopic(String name) {
        return this.hazelcastInstance.getTopic(name);
    }

    /****
     * 获取队列消息
     * @param name
     * @return
     */
    @Override
    public IQueue getQueue(String name) {
        return this.hazelcastInstance.getQueue(name);
    }


    /****
     * 获取map信息
     * @param name
     * @return
     */
    @Override
    public IMap getMap(String name) {
        return this.hazelcastInstance.getMap(name);
    }


    /***
     * 获取MultiMap信息
     * @param name
     * @return
     */
    @Override
    public MultiMap getMultiMap(String name) {
        return this.hazelcastInstance.getMultiMap(name);
    }


    /***
     * 获取复制MAP
     * @param name
     * @return
     */
    @Override
    public ReplicatedMap getReplicatedMap(String name) {
        return this.hazelcastInstance.getReplicatedMap(name);
    }


    /****
     * 获取SET信息
     * @param name
     * @return
     */
    @Override
    public ISet getSet(String name) {
        return this.hazelcastInstance.getSet(name);
    }


    /***
     * 全局计数器
     * @param name
     * @return
     */
    @Override
    public PNCounter getPNCounter(String name) {
        return this.hazelcastInstance.getPNCounter(name);
    }

    /***
     * 大计数器
     * @param name
     * @return
     */
    @Override
    public IAtomicLong getIAtomicLong(String name) {
        return this.hazelcastInstance.getCPSubsystem().getAtomicLong(name);
    }

    /***
     * 实现分布式应用
     * @param name
     * @return
     */
    @Override
    public IAtomicReference getIAtomicReference(String name) {
        return this.hazelcastInstance.getCPSubsystem().getAtomicReference(name);
    }


    /***
     * 获取分布式锁信息
     * @param name
     * @return
     */
    @Override
    public FencedLock getLock(String name) {
        return this.hazelcastInstance.getCPSubsystem().getLock(name);
    }


    /****
     * 分布式持久任务调用
     * @param name
     * @return
     */
    @Override
    public DurableExecutorService getDurableExecutorService(String name) {
        return this.hazelcastInstance.getDurableExecutorService(name);
    }


    /***
     * 获取分布式执行器
     * @param name
     * @return
     */
    @Override
    public IExecutorService getExecutorService(String name) {
        return this.hazelcastInstance.getExecutorService(name);
    }


    /***
     * 分布式延迟执行器
     * @param name
     * @return
     */
    @Override
    public IScheduledExecutorService getScheduledExecutorService(String name) {
        return this.hazelcastInstance.getScheduledExecutorService(name);
    }


    /***
     * 获取ICountDownLatch
     * @param name
     * @return
     */
    @Override
    public ICountDownLatch getCountDownLatch(String name) {
        return this.hazelcastInstance.getCPSubsystem().getCountDownLatch(name);
    }


    /***
     * 获取事务管理器
     * @param type
     * @return
     */
    @Override
    public TransactionContext getTransactionContext(TransactionOptions.TransactionType type) {
        TransactionOptions options = new TransactionOptions().setTransactionType(type);
        return this.hazelcastInstance.newTransactionContext(options);
    }


    /***
     * 销毁事件
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        for (HazelcastLoader hazelcastLoader : this.hazelcastLoaders) {
            hazelcastLoader.destroy(this.hazelcastInstance);
        }
        this.hazelcastInstance.shutdown();
    }
}